package  cn.ac.big.bigd;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;

public final class JavaWordCount {
  private static final Pattern SPACE = Pattern.compile(" ");
  public static void main(String[] args) throws Exception {
   if (args.length < 1) {
      System.err.println("Usage: JavaWordCount <file>");
      System.exit(1);
    }

    SparkConf sparkConf = new SparkConf().setAppName("count5").setMaster("spark://192.168.118.212:7077").
            setJars(new String[]{"D:\\IdeaProjects\\spark\\target\\spark-1.0.jar"});

    JavaSparkContext ctx = new JavaSparkContext(sparkConf);
//    JavaRDD<String> lines = ctx.textFile("file:///C:/home/wangfan/JavaWordCount.java", 1);

   // JavaRDD<String> lines = ctx.textFile("file:////home/wangfan/JavaWordCount.java", 1);
    JavaRDD<String> lines = ctx.textFile(args[0], 1);
    JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
      @Override
      public Iterable<String> call(String s) {
        return Arrays.asList(SPACE.split(s));
      }
    });

    JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {
      @Override
      public Tuple2<String, Integer> call(String s) {
        return new Tuple2<String, Integer>(s, 1);
      }
    });

    JavaPairRDD<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {
      @Override
      public Integer call(Integer i1, Integer i2) {
        return i1 + i2;
      }
    });

    List<Tuple2<String, Integer>> output = counts.collect();
    for (Tuple2<?,?> tuple : output) {
      System.out.println(tuple._1() + ": " + tuple._2());
    }
    /*List<Integer> l = new ArrayList<Integer>(NUM_SAMPLES);
    for (int i = 0; i < NUM_SAMPLES; i++) {
      l.add(i);
    }

    long count = ctx.parallelize(l).filter(new Function<Integer, Boolean>() {
      public Boolean call(Integer i) {
        double x = Math.random();
        double y = Math.random();
        return x*x + y*y < 1;
      }
    }).count();
    System.out.println("Pi is roughly " + 4.0 * count / NUM_SAMPLES);*/
    ctx.stop();
  }
}